跳到主要内容

Go 连接 MongoDB

初始化环境​

启动服务

mongod \
   --dbpath /var/lib/mongo \
   --logpath /var/log/mongodb/mongod.log \
   --fork

打开客户端

$ mongo

初始化仓库

// 创建数据库
use go_db;

// 创建集合
db.createCollection("student");

下载驱动

go get github.com/mongodb/mongo-go-driver

连接数据库​

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "go.mongodb.org/mongo-driver/mongo"
    "go.mongodb.org/mongo-driver/mongo/options"
)

func main() {
    var (
        client     *mongo.Client
        err        error
        db         *mongo.Database
        collection *mongo.Collection
    )

    //1.建立连接
    if client, err = mongo.Connect(context.TODO(),
        options.Client().ApplyURI("mongodb://localhost:27017").
            SetConnectTimeout(5*time.Second)); err != nil {
        fmt.Print(err)
        return
    }

    // 2.检查连接
    err = client.Ping(context.TODO(), nil)
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println("Connected to MongoDB!")

    //3.选择数据库 my_db
    db = client.Database("my_db")

    //4.选择表 my_collection
    collection = db.Collection("my_collection")
    collection = collection
}

将连接函数提取出来​

创建 util 包

import (
    "context"
    "log"

    "go.mongodb.org/mongo-driver/mongo"
    "go.mongodb.org/mongo-driver/mongo/options"
)

var mgoCli *mongo.Client

func initEngine() {
    var err error
    clientOptions := options.Client().ApplyURI("mongodb://localhost:27017")

    // 连接到 MongoDB
    mgoCli, err = mongo.Connect(context.TODO(), clientOptions)
    if err != nil {
        log.Fatal(err)
    }
    // 检查连接
    err = mgoCli.Ping(context.TODO(), nil)
    if err != nil {
        log.Fatal(err)
    }
}

func GetMgoCli() *mongo.Client {
    if mgoCli == nil {
        initEngine()
    }
    return mgoCli
}

插入一条数据​

构建几个结构体

type TimePoint struct {
    StartTime int64 `bson:"startTime"` //开始时间
    EndTime   int64 `bson:"endTime"`   //结束时间
}

type LogRecord struct {
    JobName string    `bson:"jobName"` //任务名
    Command string    `bson:"command"` //shell命令
    Err     string    `bson:"err"`     //脚本错误
    Content string    `bson:"content"` //脚本输出
    Tp      TimePoint //执行时间
}

关于这个 BSON 的 Tag 文档参考 Work with BSON

main 函数：

func main() {
    var (
        client     = util.GetMgoCli()
        err        error
        collection *mongo.Collection
        iResult    *mongo.InsertOneResult
        id         primitive.ObjectID
    )

    //2.选择数据库 my_db里的某个表
    collection = client.Database("my_db").Collection("my_collection")

    lr := &model.LogRecord{
        JobName: "PlanA",
        Command: "echo 'say hello'",
        Err:     "",
        Content: "hello",
        Tp: model.TimePoint{
            StartTime: time.Now().Unix(),
            EndTime:   time.Now().Unix() + 1000,
        },
    }

    //插入某一条数据
    if iResult, err = collection.InsertOne(context.TODO(), lr); err != nil {
        fmt.Print(err)
        return
    }

    //_id:默认生成的一个全局唯一ID
    id = iResult.InsertedID.(primitive.ObjectID)
    fmt.Println("自增ID", id.Hex())
}

检查数据：

> use my_db
// 这里的查询格式是 db.集合.find()
> db.my_collection.find()
{ "_id" : ObjectId("62b7d1503b40d8d17476c2c4"), "jobName" : "PlanA", "command" : "echo 'say hello'", "err" : "", "content" : "hello", "tp" : { "startTime" : NumberLong(1656213840), "endTime" : NumberLong(1656214840) } }

批量插入数据​

func main() {
    var (
        client     = util.GetMgoCli()
        err        error
        collection *mongo.Collection
        result     *mongo.InsertManyResult
        id         primitive.ObjectID
    )
    collection = client.Database("my_db").Collection("test")

    //批量插入
    result, err = collection.InsertMany(context.TODO(), []interface{}{
        model.LogRecord{
            JobName: "job10",
            Command: "echo 1",
            Err:     "",
            Content: "1",
            Tp: model.TimePoint{
                StartTime: time.Now().Unix(),
                EndTime:   time.Now().Unix() + 10,
            },
        },
        model.LogRecord{
            JobName: "job10",
            Command: "echo 2",
            Err:     "",
            Content: "2",
            Tp: model.TimePoint{
                StartTime: time.Now().Unix(),
                EndTime:   time.Now().Unix() + 10,
            },
        },
        model.LogRecord{
            JobName: "job10",
            Command: "echo 3",
            Err:     "",
            Content: "3",
            Tp: model.TimePoint{
                StartTime: time.Now().Unix(),
                EndTime:   time.Now().Unix() + 10,
            },
        },
        model.LogRecord{
            JobName: "job10",
            Command: "echo 4",
            Err:     "",
            Content: "4",
            Tp: model.TimePoint{
                StartTime: time.Now().Unix(),
                EndTime:   time.Now().Unix() + 10,
            },
        },
    })
    if err != nil {
        log.Fatal(err)
    }
    if result == nil {
        log.Fatal("result nil")
    }
    for _, v := range result.InsertedIDs {
        id = v.(primitive.ObjectID)
        fmt.Println("自增ID", id.Hex())
    }
}

检查 MongoDB 的内容

> db.test.find()
{ "_id" : ObjectId("62b7d8d2fde90f10033b5076"), "jobName" : "job10", "command" : "echo 1", "err" : "", "content" : "1", "tp" : { "startTime" : NumberLong(1656215762), "endTime" : NumberLong(1656215772) } }
{ "_id" : ObjectId("62b7d8d2fde90f10033b5077"), "jobName" : "job10", "command" : "echo 2", "err" : "", "content" : "2", "tp" : { "startTime" : NumberLong(1656215762), "endTime" : NumberLong(1656215772) } }
{ "_id" : ObjectId("62b7d8d2fde90f10033b5078"), "jobName" : "job10", "command" : "echo 3", "err" : "", "content" : "3", "tp" : { "startTime" : NumberLong(1656215762), "endTime" : NumberLong(1656215772) } }
{ "_id" : ObjectId("62b7d8d2fde90f10033b5079"), "jobName" : "job10", "command" : "echo 4", "err" : "", "content" : "4", "tp" : { "startTime" : NumberLong(1656215762), "endTime" : NumberLong(1656215772) } }

查询数据​

使用标准的查询方式需要先定义一个结构体用来查询，如下添加一个查询结构体：

//查询实体
type FindByJobName struct {
    JobName string `bson:"jobName"` //任务名
}

之所以需要创建一个结构体用于查询是因为原本的那个 LogRecord 里面的其它值创建时默认是零值，会影响到查询（具体看下面的注释），整个查询代码如下所示：

func main() {
    var (
        client     = util.GetMgoCli()
        err        error
        collection *mongo.Collection
        cursor     *mongo.Cursor
    )
    //2.选择数据库 my_db里的某个表
    collection = client.Database("my_db").Collection("test")

    //如果直接使用 LogRecord{JobName: "job10"}是查不到数据的，因为其他字段有初始值0或者“”
    cond := model.FindByJobName{JobName: "job10"}

    //按照jobName字段进行过滤jobName="job10",翻页参数0-2
    if cursor, err = collection.Find(context.TODO(), cond, options.Find().SetSkip(0), options.Find().SetLimit(2)); err != nil {
        fmt.Println(err)
        return
    }

    //延迟关闭游标
    defer func() {
        if err = cursor.Close(context.TODO()); err != nil {
            log.Fatal(err)
        }
    }()

    //遍历游标获取结果数据
    for cursor.Next(context.TODO()) {
        var lr model.LogRecord
        //反序列化Bson到对象
        if cursor.Decode(&lr) != nil {
            fmt.Print(err)
            return
        }
        //打印结果数据
        fmt.Println(lr)
    }
}

这里的结果遍历可以使用另外一种更方便的方式：

var results []model.LogRecord
if err = cursor.All(context.TODO(), &results); err != nil {
    log.Fatal(err)
}
for _, result := range results {
    fmt.Println(result)
}

使用 BSON 包去查询数据​

前面查询数据时需要创建一个结构体来承载查询的字段，但是 MongoDB 这种 NoSQL 数据库最大的好处就是其是动态的，如果使用上面那种写法，那又和传统的关系型数据库有什么区别呢？所以应该可以使用 BSON 包提供的方法去查询

这一块的文档参考 Work with BSON

连接 MongoDB 的 Go 驱动程序中有两大类型表示 BSON 数据：D 和 Raw。

类型 D 家族被用来简洁地构建使用本地 Go 类型的 BSON 对象。这对于构造传递给 MongoDB 的命令特别有用。D 家族包括四类:

  • D：一个 BSON 文档。这种类型应该在顺序重要的情况下使用，比如 MongoDB 命令。
  • M：一张无序的 map。它和 D 是一样的，只是它不保持顺序。
  • A：一个 BSON 数组。
  • E：D 里面的一个元素

导入以下包就可以使用了：

import "go.mongodb.org/mongo-driver/bson"

使用例如下，用于查找 name 字段与 '张三' 或 '李四' 匹配的文档：

bson.D{{
   "name",
   bson.D{{
      "$in",
      bson.A{"张三", "李四"},
   }},
}}

使用例：

func main() {
    var (
        client     = util.GetMgoCli()
        collection *mongo.Collection
        err        error
        cursor     *mongo.Cursor
    )

    //2.选择数据库 my_db里的某个表
    collection = client.Database("my_db").Collection("test")
    filter := bson.M{"jobName": "job10"}
    if cursor, err = collection.Find(context.TODO(), filter, options.Find().SetSkip(0), options.Find().SetLimit(2)); err != nil {
        log.Fatal(err)
    }
    //延迟关闭游标
    defer func() {
        if err = cursor.Close(context.TODO()); err != nil {
            log.Fatal(err)
        }
    }()

    //这里的结果遍历可以使用另外一种更方便的方式：
    var results []model.LogRecord
    if err = cursor.All(context.TODO(), &results); err != nil {
        log.Fatal(err)
    }
    for _, result := range results {
        fmt.Println(result)
    }
}

聚合查询​

func main() {
    var (
        client     = util.GetMgoCli()
        collection *mongo.Collection
        err        error
        cursor     *mongo.Cursor
    )
    //2.选择数据库 my_db里的某个表
    collection = client.Database("my_db").Collection("test")
    //filter := bson.M{"jobName": "job10"}

    //按照 jobName 分组,countJob 中存储每组的数目
    groupStage := mongo.Pipeline{{
        {"$group", bson.D{
            {"_id", "$jobName"},
            {"countJob", bson.D{
                {"$sum", 1},
            }},
        }},
    }}

    if cursor, err = collection.Aggregate(context.TODO(), groupStage); err != nil {
        log.Fatal(err)
    }

    //延迟关闭游标
    defer func() {
        if err = cursor.Close(context.TODO()); err != nil {
            log.Fatal(err)
        }
    }()

    //遍历游标
    var results []bson.M
    if err = cursor.All(context.TODO(), &results); err != nil {
        log.Fatal(err)
    }
    for _, result := range results {
        fmt.Println(result)
    }
}

打印输出：

$ go run .
map[_id:job10 countJob:4]

更新数据​

同样的，使用 mongo-go-driver 进行更新也需要建立专门用于更新的实体

在这里建立的实体中存在 Command 和 Content 两个字段，更新时需要同时对这两个字段进行赋值，否则未被赋值的字段会被更新为 Golang 的数据类型初始值。

为更新方便可以采用 bson.M{"$set": bson.M{"command": "ByBsonM",}} 来进行更新

package model
// 更新实体
type UpdateByJobName struct {
   Command string     `bson:"command"` //shell命令
   Content string     `bson:"content"` //脚本输出
}

编写具体的代码

func main() {
    var (
        client     = util.GetMgoCli()
        collection *mongo.Collection
        err        error
        uResult    *mongo.UpdateResult
    )

    //2.选择数据库 my_db里的某个表
    collection = client.Database("my_db").Collection("test")

    filter := bson.M{"jobName": "job10"}
    // update := bson.M{"$set": bson.M{"command": "ByBsonM",}}
    update := bson.M{"$set": model.UpdateByJobName{Command: "byModel", Content: "model"}}
    // update := bson.M{"$set": model.LogRecord{JobName:"job10",Command:"byModel"}}

    if uResult, err = collection.UpdateMany(context.TODO(), filter, update); err != nil {
        log.Fatal(err)
    }
    //uResult.MatchedCount表示符合过滤条件的记录数，即更新了多少条数据。
    log.Println(uResult.MatchedCount)
}

这个 $set 表示修改字段的值，例如：

bson.M{"$set": model.UpdateByJobName{Command: "byModel", Content:"model"}}

使用 $inc 可以对字段的值进行增减计算，例如下面表示对 age 的值减一。

bson.M{"$inc": bson.M{ "age": -1, }}

使用 $push 可以对该字段增加一个元素，例如下面表示对 interests 字段的元素数组增加 Golang 元素。

bson.M{"$push": bson.M{ "interests": "Golang", }}

使用 $push 也可以对该字段删除一个元素，例如下面也表示对 interests 字段的元素数组删除 Golang 元素。

bson.M{"$pull": bson.M{ "interests": "Golang", }}

删除数据​

func main() {
   var (
      client     = util.GetMgoCli()
      collection *mongo.Collection
      err        error
      uResult    *mongo.DeleteResult
      //upsertedID model.LogRecord
   )
   //2.选择数据库 my_db里的某个表
   collection = client.Database("my_db").Collection("test")
   filter := bson.M{"jobName": "job0"}
   //3.删除开始时间早于当前时间的数据
   //
   if uResult, err = collection.DeleteMany(context.TODO(), filter); err != nil {
      log.Fatal(err)
   }
   log.Println(uResult.DeletedCount)
}

带过滤条件删除：

type DeleteCond struct {
   BeforeCond TimeBeforeCond `bson:"tp.startTime"`
}

//startTime小于某时间，使用这种方式可以对想要进行的操作($set、$group等)提前定义
type TimeBeforeCond struct {
   BeforeTime int64 `bson:"$lt"`
}

func main() {
   var (
      client     = util.GetMgoCli()
      collection *mongo.Collection
      err        error
      uResult    *mongo.DeleteResult
      delCond    *DeleteCond
      //upsertedID model.LogRecord
   )
   //2.选择数据库 my_db里的某个表
   collection = client.Database("my_db").Collection("test")

   //3.删除jobName为job0的数据
   delCond = &DeleteCond{BeforeCond: TimeBeforeCond{BeforeTime: time.Now().Unix()}}
   if uResult, err = collection.DeleteMany(context.TODO(), delCond); err != nil {
      log.Fatal(err)
   }
   log.Println(uResult.DeletedCount)
}

Reference​